package com.qf.listener;

import com.qf.pojo.User;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class RabbitMQListener {

    private AtomicInteger ai = new AtomicInteger(0);

    /**
     * springboot中消费者默认确认的机制是：
     *      如果不出现异常，消费完消息后，直接确认--消息会从消息中间件中删除
     *      如果出现异常，springboot会自动进行消息消费重试
     *
     * 实际有些业务的处理成功或者失败不是依靠异常完成的，那么springboot这种方式就不合适，我们也可以手动实现确认
     *      ① 在配置文件中，开启消费端的手动确认
     *      ② 在消费代码中实现手动确认逻辑
     *
     * @param msg
     * @throws UnsupportedEncodingException
     */
//    @RabbitListener(queues = "boot.topicQue1")
    public void consumer1(Message msg, Channel channel) throws IOException {
        byte[] body = msg.getBody();
        String message = new String(body, "utf-8");

        //模拟异常
//        if (ai.incrementAndGet()<5) {
//            int i = 1 / 0;
//        }

        System.out.println("consumer1： " + message);

        //这个标记表示业务执行完后，是否正常处理业务
        boolean flag = false;
        if (flag){
            //业务正常 -- 确认消费
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
        }else{
            //业务非正常 -- 确认不消费（重试）
            if (ai.incrementAndGet()<5) {
                //入队重试
                System.out.println("消息重试.....");
                channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,true);
            }else{
                System.out.println("将消息写入到数据库的重试表中，开启一个定时任务，扫描重试表，实行定时重试");
                channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);
            }
        }

    }

//    @RabbitListener(queues = "boot.topicQue2")
    public void consumer2(Message msg) throws UnsupportedEncodingException {
        byte[] body = msg.getBody();
        String message = new String(body, "utf-8");

        System.out.println("consumer2： " + message);
    }


//    @RabbitListener(queues = "boot.topicQue1")
    public void consumer3(User user) throws UnsupportedEncodingException {

        System.out.println("consumer3： " + user);
    }


    /**=============================================*/
    //@RabbitListener(queues = "boot.proDeadMsgQue")
    public void consumer3(Message msg,Channel channel) throws IOException, InterruptedException {

        //直接拒绝,不入队，这个消息就会成为死信消息，转到死信队列中
        Thread.sleep(5000);
//        channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
    }

    //@RabbitListener(queues = "dlQue")
    public void consumer4(Message message){
        System.out.println("检查订单状态，检查时间为："+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
    }

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 模拟重复消费消息
     * @param message
     */
    @RabbitListener(queues = "boot.topicQue2")
    public void consumer5(Message message,Channel channel) throws IOException {
        /*
            如果这个消费者是一个查询业务 -- 重复消费没有问题
            如果这个消费者是一个删除业务 -- 重复消费没有问题
            如果这个消费者是一个更新业务 -- 重复消费可能问题
                更新操作是一个设置型的： update xx set yy=1000 where x=xx  没有问题
                更新操作是一个计算型的： update xx set yy=yy+100 where x=xx  有问题
                    ①查询这个消息id的操作是否有执行过，如果有执行过，不再执行，如果没有执行过，需要执行
                        如何判断id有没有执行过？？？ redis---> setnx (msgId,xxx)
                    ② 在数据库的字段上利用状态机原理，实现数据状态的判断
                        update xx set yy=yy+100,status=已修改 where x=xx and status=未修改过
            如果这个消费者是一个新增业务 -- 重复消费一定有问题
                ① redis加锁，利用业务id作为锁对象
                    根据业务id查询，如果能查到，不再执行；如果查不到，执行新增操作
                ② 利用数据库的悲观锁实现
                    根据业务id查询，select xx from xxx where id= 业务id for update;
                    如果能查到，不再执行，如果查不到，执行新增操作
                    上述查询操作和之后的判断、新增逻辑都是在一个事务中，事务占有锁的时间太长了，数据库的性能会降低
                ③ 利用唯一约束 （id不允许重复，唯一约束的键值不允许重复）
         */
        String messageId = message.getMessageProperties().getMessageId();

        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(messageId, "1");

        if (aBoolean) {
            byte[] body = message.getBody();
            String msg = new String(body, "utf-8");
            System.out.println("msgId: " + messageId);
            System.out.println("msg: " + msg);
        }

        //确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}
